package com.ahcloud.edas.common.kafka;

import com.ahcloud.edas.common.kafka.client.KafkaMessageClient;
import com.ahcloud.edas.common.kafka.listener.MessageListener;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.Objects;
import java.util.Properties;

/**
 * @program: edas-common
 * @description:
 * @author: YuKai Fan
 * @create: 2024/4/20 09:44
 **/
public abstract class AbstractConsumer<K, V> extends BaseConsumer implements AckManually<K, V>, KafkaConsumerInner {

    private static final Logger log = LoggerFactory.getLogger(AbstractConsumer.class);
    private AckMode ackMode = AckMode.AT_LEAST_ONCE;

    private final KafkaMessageClient client;
    private final KafkaConsumer<K, V> consumer;

    private Duration pollTime = Duration.of(10, ChronoUnit.SECONDS);

    protected AbstractConsumer(String consumerGroup, KafkaMessageClient client) {
        this(consumerGroup, null, client);
    }

    protected AbstractConsumer(String consumerGroup, Properties properties, KafkaMessageClient client) {
        super(client.getEnv(), consumerGroup);
        if (getMessageListenerInner() == null) {
            throw new RuntimeException("The consumer MessageListener is null");
        }
        if (Objects.isNull(properties)) {
            properties = new Properties();
        }
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
        this.consumer = client.createConsumer(properties);
        this.client = client;

        // 订阅topic
        this.consumer.subscribe(Collections.singleton(getTopic() + LINK + getSuffix()));

        boolean registerOK = KafkaClientInstance.getInstance().registerConsumer(consumerGroup, this);
        if (!registerOK) {
            throw new RuntimeException("The consumer group[" + consumerGroup
                    + "] has been created before, specify another name please.");
        }
    }

    @Override
    public void run() {
        log.info("{} service started", this.getServiceName());
        while (!this.isStopped()) {
            try {
                this.pullMessage();
            } catch (Exception e) {
                log.error("Pull Message Run Method exception", e);
            }
        }
    }

    @Override
    public void pullMessage() {
        // 如果当前没有数据可以消费，那么会等待 pollTime，默认10s
        ConsumerRecords<K, V> result = consumer.poll(this.pollTime);
        if (result != null && !result.isEmpty()) {
            if (ackMode == AckMode.AT_MOST_ONCE) {
                ack(result);
            }
            getMessageListenerInner().received(consumer, result);
            if (ackMode == AckMode.AT_LEAST_ONCE) {
                ack(result);
            }
        }
    }

    @Override
    public void consumerStart() {
        super.start();
    }

    @Override
    public void consumerShutdown() {
        this.consumer.close();
        super.shutdown();
    }

    @Override
    public void ack(ConsumerRecords<K, V> messages) {
        // 采用异步手动提交ack，因为即使异步提交失败，相当于变相的重试，只要客户端保证幂等性即可
        this.consumer.commitAsync();
    }

    @Override
    public Properties getProperties() {
        return this.client.getConsumerProperties();
    }

    protected abstract MessageListener<K, V> getMessageListenerInner();

    public Duration getPollTime() {
        return pollTime;
    }

    public void setPollTime(Duration pollTime) {
        this.pollTime = pollTime;
    }

    public AckMode getAckMode() {
        return ackMode;
    }

    public void setAckMode(AckMode ackMode) {
        if (ackMode == null) {
            return;
        }
        this.ackMode = ackMode;
    }

    protected abstract String getTopic();
}
